package loadcount

import (
	"bufio"
	"fmt"
	"io"
	"os"
	"strconv"
	"strings"

	"gitee.com/smewl13/learn/utils"
)

type Options struct {
	StartTime string
	EndTime   string
	Cmd       string
	SqlFile   string
	Conf      *utils.Yaml
}

func LoadCount(options *Options) {
	hostName, err := os.Hostname()
	if err != nil {
		utils.Error.Printf("ERROR: 未获取到主机名: %s", err)
		hostName = "localhost"
	}

	db := ConnectDB(&options.Conf.Mysql)
	switch options.Cmd {
	case "load":
		db.rollingCount(hostName)
		db.merageCount(hostName)
	case "query":
		rollingResult := db.RollingQuery(options.StartTime, options.EndTime)
		fmt.Println(rollingResult)
		merageResult := db.MerageQuery(options.StartTime, options.EndTime)
		fmt.Println(merageResult)
	default:
		fmt.Printf("命令错误: %s\n", options.Cmd)
	}
}

// func (db *DBConnect) query(options *Options){
// 	var sql string
// 	if options.SqlFile == ""{
// 		sql =
// 	}
// }

func (db *DBConnect) rollingCount(hostName string) {
	rollingFile := "/opt/cloudera/parcels/load/logs/rolling.log"
	file, err := os.Open(rollingFile)
	defer file.Close()
	if err != nil {
		utils.Error.Printf("ERROR: 打开文件%s失败: %s\n", rollingFile, err)
		return
	}

	delSql := fmt.Sprintf("delete from rolling where time >= DATE_SUB(curdate(), INTERVAL 0 DAY) and host = '%s';", hostName)
	db.DbOperator(delSql)

	reader := bufio.NewReader(file)
	for {
		lineBytes, _, err := reader.ReadLine()
		if err == io.EOF {
			utils.Info.Printf("INFO: 源文件%s读取结束\n", rollingFile)
			break
		}
		lineStr := strings.Replace(string(lineBytes), ",", "", -1)
		column := strings.Split(lineStr, " ")
		if len(column) < 32 {
			continue
		}
		if column[7] == "数据行队列当前大小:" && column[6] != "总计" {
			dTime := strings.Join(column[:2], " ")
			tableName := TableCode[strings.Split(column[6], "_")[0]]
			receive, err := strconv.ParseInt(strings.Replace(column[13], "行", "", -1), 10, 64)
			if err != nil {
				utils.Error.Printf("接收数据量字段不是整型数值(14列)。原始数据: %s\n", lineStr)
				continue
			}
			processed, err := strconv.ParseInt(column[16], 10, 64)
			if err != nil {
				utils.Error.Printf("已处理的行数字段不是整型数值(17列)。原始数据: %s\n", lineStr)
				continue
			}
			success, err := strconv.ParseInt(column[19], 10, 64)
			if err != nil {
				utils.Error.Printf("处理成功行数字段不是整型数值(20列)。原始数据: %s\n", lineStr)
				continue
			}
			failed, err := strconv.ParseInt(column[21], 10, 64)
			if err != nil {
				utils.Error.Printf("处理失败行数字段不是整型数值(22列)。原始数据: %s\n", lineStr)
				continue
			}
			data := &Rolling{
				Time:      dTime,
				HostName:  hostName,
				TableName: tableName,
				Receive:   receive,
				Processed: processed,
				Success:   success,
				Failed:    failed,
			}
			db.InsertRolling(data)
		}
	}
}

func (db *DBConnect) merageCount(hostName string) {
	merageFile := "/opt/cloudera/parcels/load/logs/rolling_merge.log"
	file, err := os.Open(merageFile)
	defer file.Close()
	if err != nil {
		utils.Error.Printf("ERROR: 打开文件%s失败: %s\n", merageFile, err)
		return
	}

	delSql := fmt.Sprintf("delete from merage where time >= DATE_SUB(curdate(), INTERVAL 0 DAY) and host = '%s';", hostName)
	db.DbOperator(delSql)

	reader := bufio.NewReader(file)
	for {
		lineBytes, _, err := reader.ReadLine()
		if err == io.EOF {
			utils.Info.Printf("INFO: 源文件%s读取结束\n", merageFile)
			break
		}
		lineStr := strings.Replace(string(lineBytes), ",", "", -1)
		column := strings.Split(lineStr, " ")
		if len(column) < 32 {
			continue
		}
		if column[18] == "数据条数" && column[8] == "表分区:" {
			dTime := strings.Join(column[:2], " ")
			tableName := strings.Split(column[9], "_V")[0]
			recordNum, err := strconv.ParseInt(column[19], 10, 64)
			if err != nil {
				utils.Error.Printf("已处理的行数字段不是整型数值(17列)。原始数据: %s\n", lineStr)
				continue
			}
			failed, err := strconv.ParseInt(column[21], 10, 64)
			if err != nil {
				utils.Error.Printf("处理失败行数字段不是整型数值(22列)。原始数据: %s\n", lineStr)
				continue
			}
			data := &Merage{
				Time:      dTime,
				HostName:  hostName,
				TableName: tableName,
				RecordNum: recordNum,
				Failed:    failed,
			}
			db.InsertMerage(data)
		}
	}
}
